/*-------------------------------------------------------------------------
 *
 * origin.c
 *	  逻辑复制进度跟踪支持。
 *
 * Copyright (c) 2013-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/logical/origin.c
 *
 * NOTES
 *
 * 此文件提供以下内容：
 * * 为复制设置命名节点的基础设施
 * * 一种高效存储和持久化复制进度的机制。
 *
 * 复制源由描述性、用户定义的外部名称和一个简短的，因此节省
 * 空间的内部 2 字节名称组成。这种拆分的存在是因为复制源必须存
 * 储在 WAL 和共享内存中，而长描述符效率低下。现在只使用 2 字节
 * 作为复制源的内部 ID，因为不太可能在一个复制设置中有超过 65k
 * 个节点；并且只使用两字节使我们能够更节省空间。
 *
 * 复制进度在一个共享内存表 (ReplicationState) 中跟踪，该表在每
 * 次检查点时转储到磁盘。该表中的条目（“插槽”）由内部 ID 识别。
 * 之所以这样是因为它允许在崩溃恢复期间增加复制进度。为了能做到
 * 这一点，我们在提交记录中存储了事务的原始 LSN（来自原始系统）。
 * 这允许在崩溃恢复后恢复精确的重放状态；无需同步提交。允许逻辑
 * 复制使用异步提交一般对性能有利，但特别重要，因为它允许单线程
 * 的重放过程跟上具有多个并发生成更改的源。出于效率和简单性，后
 * 端可以设置一个复制源，之后作为后端生成的更改的源，直到再次重置。
 *
 * 该基础设施旨在与逻辑解码协同使用。当从远程系统进行重放时，
 * 配置的源会提供给输出插件，从而防止复制循环和其他过滤。
 *
 * 这里有多个级别的锁定：
 *
 * * 创建和删除复制源需要对 pg_replication_slot 的独占锁定，锁定
 *	 持续期间。这使我们能够安全、无冲突地使用脏快照分配新源。
 *
 * * 在创建内存中的复制进度插槽时，必须独占拥有 ReplicationOrigin
 *	 LWLock；在遍历复制进度时，必须持有共享锁，进展尚未设置为会话
 *	 的复制源的单个后端的复制进度时也是如此。
 *
 * * 在操作或查看复制进度插槽的 remote_lsn 和 local_lsn 字段时，必
 *	 须持有该插槽的 lwlock。这主要是因为我们不假设 8 字节的写入
 *	 （LSN）在所有平台上都是原子的，但它也简化了远程和本地 LSN 之
 *	 间的内存排序问题。我们使用 lwlock 而不是自旋锁，因此在 WAL 写入
 *	 期间持有锁的影响要小得多（参见 AdvanceReplicationProgress）。
 *
 * ---------------------------------------------------------------------------
 */

#include "postgres.h"

#include <unistd.h>
#include <sys/stat.h>

#include "access/genam.h"
#include "access/htup_details.h"
#include "access/table.h"
#include "access/xact.h"
#include "access/xloginsert.h"
#include "catalog/catalog.h"
#include "catalog/indexing.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "nodes/execnodes.h"
#include "pgstat.h"
#include "replication/logical.h"
#include "replication/origin.h"
#include "storage/condition_variable.h"
#include "storage/copydir.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/lmgr.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/pg_lsn.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"

/*
 * 单个远程节点的重放进度。
 */
typedef struct ReplicationState
{
	/*
	 * 远程节点的本地标识符。
	 */
	RepOriginId roident;

	/*
	 * 远程端最新提交的位置。
	 */
	XLogRecPtr	remote_lsn;

	/*
	 * 记住提交记录的本地 lsn，以便我们可以在检查点期间进行 XLogFlush()
	 * 以确保提交记录实际上是安全的在磁盘上。
	 */
	XLogRecPtr	local_lsn;

	/*
	 * 获取槽位的后端的 PID，如果没有则为 0。
	 */
	int			acquired_by;

	/*
	 * 当 acquired_by 变化时发出信号的条件变量。
	 */
	ConditionVariable origin_cv;

	/*
	 * 保护 remote_lsn 和 local_lsn 的锁。
	 */
	LWLock		lock;
} ReplicationState;

/*
 * 磁盘上的 ReplicationState 版本。
 */
typedef struct ReplicationStateOnDisk
{
	RepOriginId roident;
	XLogRecPtr	remote_lsn;
} ReplicationStateOnDisk;


typedef struct ReplicationStateCtl
{
	/* 用于每个来源的 LWLocks 的批次 */
	int			tranche_id;
	/* 最大复制槽的数组长度 */
	ReplicationState states[FLEXIBLE_ARRAY_MEMBER];
} ReplicationStateCtl;

/* 外部变量 */
RepOriginId replorigin_session_origin = InvalidRepOriginId; /* 假定身份 */
XLogRecPtr	replorigin_session_origin_lsn = InvalidXLogRecPtr;
TimestampTz replorigin_session_origin_timestamp = 0;

/*
 * 指向大小为 max_replication_slots 的共享内存阵列中复制状态的基本地址。
 *
 * XXX：我们是否应该使用一个单独的变量来定义此大小，而不是 
 * max_replication_slots？
 */
static ReplicationState *replication_states;

/*
 * 实际的共享内存块（replication_states[] 现在是这个的一部分）。
 */
static ReplicationStateCtl *replication_states_ctl;

/*
 * 后端本地，来自 ReplicationState 的缓存元素，用于后端重放远程提交，
 * 因此我们不必搜索 ReplicationState 获取后端的当前 RepOriginId。
 */
static ReplicationState *session_replication_state = NULL;

/* 磁盘文件的魔法。 */
#define REPLICATION_STATE_MAGIC ((uint32) 0x1257DADE)

static void fc_replorigin_check_prerequisites(bool fc_check_slots, bool fc_recoveryOK)
{
	if (fc_check_slots && max_replication_slots == 0)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot query or manipulate replication origin when max_replication_slots = 0")));

	if (!fc_recoveryOK && RecoveryInProgress())
		ereport(ERROR,
				(errcode(ERRCODE_READ_ONLY_SQL_TRANSACTION),
				 errmsg("cannot manipulate replication origins during recovery")));
}


/* ---------------------------------------------------------------------------
 * 处理复制来源本身的函数。
 * ---------------------------------------------------------------------------
 */

/*
 * 检查通过名称识别的持久复制来源。
 *
 * 如果节点尚未知且 missing_ok 为真，则返回 InvalidOid。
 */
RepOriginId
replorigin_by_name(const char *fc_roname, bool fc_missing_ok)
{
	Form_pg_replication_origin fc_ident;
	Oid			fc_roident = InvalidOid;
	HeapTuple	fc_tuple;
	Datum		fc_roname_d;

	fc_roname_d = CStringGetTextDatum(fc_roname);

	fc_tuple = SearchSysCache1(REPLORIGNAME, fc_roname_d);
	if (HeapTupleIsValid(fc_tuple))
	{
		fc_ident = (Form_pg_replication_origin) GETSTRUCT(fc_tuple);
		fc_roident = fc_ident->roident;
		ReleaseSysCache(fc_tuple);
	}
	else if (!fc_missing_ok)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("replication origin \"%s\" does not exist",
						fc_roname)));

	return fc_roident;
}


#ifdef FDDM
//存量数据同步创建origin id：从10000开始
//增量通过replorigin_create限制最大为9999，避免存量数据产生update_origin_change冲突<sep>//存量同步没有找到可用的<sep>//非存量origin id ，即增量、普通逻辑复制、多活逻辑复制，origin超过MIN_ORIGIN_ID_FOR_TMP<sep>//报错
RepOriginId fdd_replorigin_create_for_table_sync(const char *fc_roname)
{
	Oid			fc_roident;
	HeapTuple	fc_tuple = NULL;
	Relation	fc_rel;
	Datum		fc_roname_d;
	SnapshotData fc_SnapshotDirty;
	SysScanDesc fc_scan;
	ScanKeyData fc_key;
	Oid fc_start_origin = MIN_ORIGIN_ID_FOR_TMP;


	fc_roname_d = CStringGetTextDatum(fc_roname);

	Assert(IsTransactionState());

	/*
	 * 我们需要数字复制来源宽度为 16 位，因此不能依赖正常 oid 分配。
	 * 相反，我们简单扫描 pg_replication_origin 以找到第一个未使用的 id。
	 * 这效率不高，但这应该是一个相对不频繁的操作 - 当需要更快时，
	 * 我们可以更轻松地在这方面花费更多的代码。
	 *
	 * 我们通过在搜索期间对表采取独占锁（允许读取）来处理并发。
	 * 由于我们使用“脏快照”，我们可以读取其他正在进行会话写入的行，
	 * 即使它们在正常快照中是不可见的。由于独占锁的存在，在我们检查时
	 * 不会出现新行的危险。
	 */
	InitDirtySnapshot(fc_SnapshotDirty);

	fc_rel = table_open(ReplicationOriginRelationId, ExclusiveLock);

	for (fc_roident = fc_start_origin; fc_roident < PG_UINT16_MAX; fc_roident++)
	{
		bool		fc_nulls[Natts_pg_replication_origin];
		Datum		fc_values[Natts_pg_replication_origin];
		bool		fc_collides;

		CHECK_FOR_INTERRUPTS();
		
		ScanKeyInit(&fc_key,
					Anum_pg_replication_origin_roident,
					BTEqualStrategyNumber, F_OIDEQ,
					ObjectIdGetDatum(fc_roident));

		fc_scan = systable_beginscan(fc_rel, ReplicationOriginIdentIndex,
								  true /* indexOK */ ,
								  &fc_SnapshotDirty,
								  1, &fc_key);

		fc_collides = HeapTupleIsValid(systable_getnext(fc_scan));

		systable_endscan(fc_scan);

		if (!fc_collides)
		{
			/*
			 * 好的，找到了未使用的 roident，插入新行并进行 CCI，
			 * 以便我们的调用方可以根据需要查找它。
			 */
			memset(&fc_nulls, 0, sizeof(fc_nulls));

			fc_values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(fc_roident);
			fc_values[Anum_pg_replication_origin_roname - 1] = fc_roname_d;

			fc_tuple = heap_form_tuple(RelationGetDescr(fc_rel), fc_values, fc_nulls);
			CatalogTupleInsert(fc_rel, fc_tuple);
			CommandCounterIncrement();
			break;
		}
	}

	/* 现在再次释放锁，	*/
	table_close(fc_rel, ExclusiveLock);
	if (fc_tuple == NULL)
	{
		//存量同步没有找到可用的
		ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("could not find free replication origin ID during table synchronization.")));
	}

	heap_freetuple(fc_tuple);
	return fc_roident;
}
#endif

/*
 * 创建一个复制来源。
 *
 * 需要在事务中调用。
 */
RepOriginId
replorigin_create(const char *fc_roname)
{
	Oid			fc_roident;
	HeapTuple	fc_tuple = NULL;
	Relation	fc_rel;
	Datum		fc_roname_d;
	SnapshotData fc_SnapshotDirty;
	SysScanDesc fc_scan;
	ScanKeyData fc_key;

	fc_roname_d = CStringGetTextDatum(fc_roname);

	Assert(IsTransactionState());

	/*
	 * 我们需要数字复制来源宽度为 16 位，因此不能依赖正常 oid 分配。
	 * 相反，我们简单扫描 pg_replication_origin 以找到第一个未使用的 id。
	 * 这效率不高，但这应该是一个相对不频繁的操作 - 当需要更快时，
	 * 我们可以更轻松地在这方面花费更多的代码。
	 *
	 * 我们通过在搜索期间对表采取独占锁（允许读取）来处理并发。
	 * 由于我们使用“脏快照”，我们可以读取其他正在进行会话写入的行，
	 * 即使它们在正常快照中是不可见的。由于独占锁的存在，在我们检查时
	 * 不会出现新行的危险。
	 */
	InitDirtySnapshot(fc_SnapshotDirty);

	fc_rel = table_open(ReplicationOriginRelationId, ExclusiveLock);

	for (fc_roident = InvalidOid + 1; fc_roident < PG_UINT16_MAX; fc_roident++)
	{
		bool		fc_nulls[Natts_pg_replication_origin];
		Datum		fc_values[Natts_pg_replication_origin];
		bool		fc_collides;

		CHECK_FOR_INTERRUPTS();
#ifdef FDDM
		if (fc_roident >= MIN_ORIGIN_ID_FOR_TMP)
		{
			//非存量origin id ，即增量、普通逻辑复制、多活逻辑复制，origin超过MIN_ORIGIN_ID_FOR_TMP
			//报错
			ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("could not find free replication origin ID (capacity limit:%d) for normal slot.", MIN_ORIGIN_ID_FOR_TMP + 1)));
		}
#endif

		ScanKeyInit(&fc_key,
					Anum_pg_replication_origin_roident,
					BTEqualStrategyNumber, F_OIDEQ,
					ObjectIdGetDatum(fc_roident));

		fc_scan = systable_beginscan(fc_rel, ReplicationOriginIdentIndex,
								  true /* indexOK */ ,
								  &fc_SnapshotDirty,
								  1, &fc_key);

		fc_collides = HeapTupleIsValid(systable_getnext(fc_scan));

		systable_endscan(fc_scan);

		if (!fc_collides)
		{
			/*
			 * 好的，找到了未使用的 roident，插入新行并进行 CCI，
			 * 以便我们的调用方可以根据需要查找它。
			 */
			memset(&fc_nulls, 0, sizeof(fc_nulls));

			fc_values[Anum_pg_replication_origin_roident - 1] = ObjectIdGetDatum(fc_roident);
			fc_values[Anum_pg_replication_origin_roname - 1] = fc_roname_d;

			fc_tuple = heap_form_tuple(RelationGetDescr(fc_rel), fc_values, fc_nulls);
			CatalogTupleInsert(fc_rel, fc_tuple);
			CommandCounterIncrement();
			break;
		}
	}

	/* 现在再次释放锁，	*/
	table_close(fc_rel, ExclusiveLock);

	if (fc_tuple == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
				 errmsg("could not find free replication origin ID")));

	heap_freetuple(fc_tuple);
	return fc_roident;
}

/*
 * 删除复制源的辅助函数。
 */
static void fc_replorigin_drop_guts(Relation fc_rel, RepOriginId fc_roident, bool fc_nowait)
{
	HeapTuple	fc_tuple;
	int			fc_i;

	/*
	 * 首先，清理槽状态信息，如果有任何匹配的槽。
	 */
restart:
	fc_tuple = NULL;
	LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);

	for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
	{
		ReplicationState *fc_state = &replication_states[fc_i];

		if (fc_state->roident == fc_roident)
		{
			/* 找到我们的槽，它忙吗？ */
			if (fc_state->acquired_by != 0)
			{
				ConditionVariable *fc_cv;

				if (fc_nowait)
					ereport(ERROR,
							(errcode(ERRCODE_OBJECT_IN_USE),
							 errmsg("could not drop replication origin with ID %d, in use by PID %d",
									fc_state->roident,
									fc_state->acquired_by)));

				/*
				 * 我们必须等待然后重试。由于我们不知道哪个CV
				 * 直到这个点才等待，因此我们不能随便使用
				 * ConditionVariablePrepareToSleep（在这里调用它是
				 * 错误的，因为这样我们可能会错过信号）；只需
				 * 直接使用ConditionVariableSleep。
				 */
				fc_cv = &fc_state->origin_cv;

				LWLockRelease(ReplicationOriginLock);

				ConditionVariableSleep(fc_cv, WAIT_EVENT_REPLICATION_ORIGIN_DROP);
				goto restart;
			}

			/* 首先制作一个WAL日志条目 */
			{
				xl_replorigin_drop fc_xlrec;

				fc_xlrec.node_id = fc_roident;
				XLogBeginInsert();
				XLogRegisterData((char *) (&fc_xlrec), sizeof(fc_xlrec));
				XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_DROP);
			}

			/* 然后清除内存中的槽 */
			fc_state->roident = InvalidRepOriginId;
			fc_state->remote_lsn = InvalidXLogRecPtr;
			fc_state->local_lsn = InvalidXLogRecPtr;
			break;
		}
	}
	LWLockRelease(ReplicationOriginLock);
	ConditionVariableCancelSleep();

	/*
	 * 现在，我们可以删除目录条目。
	 */
	fc_tuple = SearchSysCache1(REPLORIGIDENT, ObjectIdGetDatum(fc_roident));
	if (!HeapTupleIsValid(fc_tuple))
		elog(ERROR, "cache lookup failed for replication origin with ID %d",
			 fc_roident);

	CatalogTupleDelete(fc_rel, &fc_tuple->t_self);
	ReleaseSysCache(fc_tuple);

	CommandCounterIncrement();
}

/*
 * 按名称删除复制源。
 *
 * 需要在事务中调用。
 */
void replorigin_drop_by_name(const char *fc_name, bool fc_missing_ok, bool fc_nowait)
{
	RepOriginId fc_roident;
	Relation	fc_rel;

	Assert(IsTransactionState());

	/*
	 * 为了防止并发删除，我们在pg_replication_origin上持有ExclusiveLock
	 * 直到事务提交。
	 *
	 * XXX 我们可以通过在特定源上获取锁来优化此操作，
	 * 如果需要使用LockSharedObject。但是，为此，我们首先
	 * 需要在ReplicationOriginRelationId上获取锁，获得origin_id，
	 * 然后锁定特定的源，最后重新检查该源是否仍然存在。
	 */
	fc_rel = table_open(ReplicationOriginRelationId, ExclusiveLock);

	fc_roident = replorigin_by_name(fc_name, fc_missing_ok);

	if (OidIsValid(fc_roident))
		fc_replorigin_drop_guts(fc_rel, fc_roident, fc_nowait);

	/* 我们在pg_replication_origin上保持锁直到提交 */
	table_close(fc_rel, NoLock);
}

/*
 * 通过其oid查找复制源并返回名称。
 *
 * 外部名称在调用上下文中分配。
 *
 * 如果已知源则返回true，否则返回false。
 */
bool replorigin_by_oid(RepOriginId fc_roident, bool fc_missing_ok, char **fc_roname)
{
	HeapTuple	fc_tuple;
	Form_pg_replication_origin fc_ric;

	Assert(OidIsValid((Oid) fc_roident));
	Assert(fc_roident != InvalidRepOriginId);
	Assert(fc_roident != DoNotReplicateId);

	fc_tuple = SearchSysCache1(REPLORIGIDENT,
							ObjectIdGetDatum((Oid) fc_roident));

	if (HeapTupleIsValid(fc_tuple))
	{
		fc_ric = (Form_pg_replication_origin) GETSTRUCT(fc_tuple);
		*fc_roname = text_to_cstring(&fc_ric->roname);
		ReleaseSysCache(fc_tuple);

		return true;
	}
	else
	{
		*fc_roname = NULL;

		if (!fc_missing_ok)
			ereport(ERROR,
					(errcode(ERRCODE_UNDEFINED_OBJECT),
					 errmsg("replication origin with ID %d does not exist",
							fc_roident)));

		return false;
	}
}


/* ---------------------------------------------------------------------------
 * 处理复制进度的函数。
 * ---------------------------------------------------------------------------
 */

Size ReplicationOriginShmemSize(void)
{
	Size		fc_size = 0;

	/*
	 * XXX: max_replication_slots 可能是错误的选择，因为在这里
	 * 我们保留*远程*事务的重放状态。但现在似乎
	 * 重新使用它是足够的，而不是引入一个单独的GUC。
	 */
	if (max_replication_slots == 0)
		return fc_size;

	fc_size = add_size(fc_size, offsetof(ReplicationStateCtl, states));

	fc_size = add_size(fc_size,
					mul_size(max_replication_slots, sizeof(ReplicationState)));
	return fc_size;
}

void ReplicationOriginShmemInit(void)
{
	bool		fc_found;

	if (max_replication_slots == 0)
		return;

	replication_states_ctl = (ReplicationStateCtl *)
		ShmemInitStruct("ReplicationOriginState",
						ReplicationOriginShmemSize(),
						&fc_found);
	replication_states = replication_states_ctl->states;

	if (!fc_found)
	{
		int			fc_i;

		MemSet(replication_states_ctl, 0, ReplicationOriginShmemSize());

		replication_states_ctl->tranche_id = LWTRANCHE_REPLICATION_ORIGIN_STATE;

		for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
		{
			LWLockInitialize(&replication_states[fc_i].lock,
							 replication_states_ctl->tranche_id);
			ConditionVariableInit(&replication_states[fc_i].origin_cv);
		}
	}
}

/* ---------------------------------------------------------------------------
 * 执行每个复制源进度的检查点，关于已重播的 remote_lsn。确保我们在
 * 检查点（local_lsn）中提到的所有事务实际上都已在磁盘上。若事务最初是异步提交的，
 * 这可能还不是情况。
 *
 * 我们以以下格式存储检查点：
 * +-------+------------------------+------------------+-----+--------+
 * | MAGIC | ReplicationStateOnDisk | struct Replic... | ... | CRC32C | EOF
 * +-------+------------------------+------------------+-----+--------+
 *
 * 所以它只是魔术值，后面跟着静态大小的
 * ReplicationStateOnDisk 结构体。请注意，ReplicationState 的最大数量
 * 由 max_replication_slots 决定。
 * ---------------------------------------------------------------------------
 */
void CheckPointReplicationOrigin(void)
{
	const char *fc_tmppath = "pg_logical/replorigin_checkpoint.tmp";
	const char *fc_path = "pg_logical/replorigin_checkpoint";
	int			fc_tmpfd;
	int			fc_i;
	uint32		fc_magic = REPLICATION_STATE_MAGIC;
	pg_crc32c	fc_crc;

	if (max_replication_slots == 0)
		return;

	INIT_CRC32C(fc_crc);

	/* 确保没有旧的临时文件存在 */
	if (unlink(fc_tmppath) < 0 && errno != ENOENT)
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not remove file \"%s\": %m",
						fc_tmppath)));

	/*
	 * 其他后端不能同时执行此操作；一次只能进行一个检查点。
	 */
	fc_tmpfd = OpenTransientFile(fc_tmppath,
							  O_CREAT | O_EXCL | O_WRONLY | PG_BINARY);
	if (fc_tmpfd < 0)
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not create file \"%s\": %m",
						fc_tmppath)));

	/* 写入魔术值 */
	errno = 0;
	if ((write(fc_tmpfd, &fc_magic, sizeof(fc_magic))) != sizeof(fc_magic))
	{
		/* 如果写入未设置 errno，假设问题是没有磁盘空间 */
		if (errno == 0)
			errno = ENOSPC;
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not write to file \"%s\": %m",
						fc_tmppath)));
	}
	COMP_CRC32C(fc_crc, &fc_magic, sizeof(fc_magic));

	/* 防止并发创建/删除 */
	LWLockAcquire(ReplicationOriginLock, LW_SHARED);

	/* 写入实际数据 */
	for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
	{
		ReplicationStateOnDisk fc_disk_state;
		ReplicationState *fc_curstate = &replication_states[fc_i];
		XLogRecPtr	fc_local_lsn;

		if (fc_curstate->roident == InvalidRepOriginId)
			continue;

		/* 清零，以避免未初始化的填充字节 */
		memset(&fc_disk_state, 0, sizeof(fc_disk_state));

		LWLockAcquire(&fc_curstate->lock, LW_SHARED);

		fc_disk_state.roident = fc_curstate->roident;

		fc_disk_state.remote_lsn = fc_curstate->remote_lsn;
		fc_local_lsn = fc_curstate->local_lsn;

		LWLockRelease(&fc_curstate->lock);

		/* 确保我们只写入一个持久的提交 */
		XLogFlush(fc_local_lsn);

		errno = 0;
		if ((write(fc_tmpfd, &fc_disk_state, sizeof(fc_disk_state))) !=
			sizeof(fc_disk_state))
		{
			/* 如果写入未设置 errno，假设问题是没有磁盘空间 */
			if (errno == 0)
				errno = ENOSPC;
			ereport(PANIC,
					(errcode_for_file_access(),
					 errmsg("could not write to file \"%s\": %m",
							fc_tmppath)));
		}

		COMP_CRC32C(fc_crc, &fc_disk_state, sizeof(fc_disk_state));
	}

	LWLockRelease(ReplicationOriginLock);

	/* 写出 CRC */
	FIN_CRC32C(fc_crc);
	errno = 0;
	if ((write(fc_tmpfd, &fc_crc, sizeof(fc_crc))) != sizeof(fc_crc))
	{
		/* 如果写入未设置 errno，假设问题是没有磁盘空间 */
		if (errno == 0)
			errno = ENOSPC;
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not write to file \"%s\": %m",
						fc_tmppath)));
	}

	if (CloseTransientFile(fc_tmpfd) != 0)
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not close file \"%s\": %m",
						fc_tmppath)));

	/* fsync，重命名为永久文件，fsync 文件和目录 */
	durable_rename(fc_tmppath, fc_path, PANIC);
}

/*
 * 从之前由 CheckPointReplicationOrigin 保存的检查点数据恢复复制重播状态。
 *
 * 这只需要在启动时调用，而不是在恢复过程中每次检查点读取期间
 * （例如，在 HS 或从基础备份进行 PITR）之后。之后的所有状态
 * 都可以通过查看提交记录恢复。
 */
void StartupReplicationOrigin(void)
{
	const char *fc_path = "pg_logical/replorigin_checkpoint";
	int			fc_fd;
	int			fc_readBytes;
	uint32		fc_magic = REPLICATION_STATE_MAGIC;
	int			fc_last_state = 0;
	pg_crc32c	fc_file_crc;
	pg_crc32c	fc_crc;

	/* 不想覆盖已经存在的状态 */
#ifdef USE_ASSERT_CHECKING
	static bool fc_already_started = false;

	Assert(!fc_already_started);
	fc_already_started = true;
#endif

	if (max_replication_slots == 0)
		return;

	INIT_CRC32C(fc_crc);

	elog(DEBUG2, "starting up replication origin progress state");

	fc_fd = OpenTransientFile(fc_path, O_RDONLY | PG_BINARY);

	/*
	 * 可能在上一次运行时 max_replication_slots == 0，或者我们刚刚
	 * 启动了一个备用。
	 */
	if (fc_fd < 0 && errno == ENOENT)
		return;
	else if (fc_fd < 0)
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not open file \"%s\": %m",
						fc_path)));

	/* 验证魔术值，即使没有任何活动也要写入 */
	fc_readBytes = read(fc_fd, &fc_magic, sizeof(fc_magic));
	if (fc_readBytes != sizeof(fc_magic))
	{
		if (fc_readBytes < 0)
			ereport(PANIC,
					(errcode_for_file_access(),
					 errmsg("could not read file \"%s\": %m",
							fc_path)));
		else
			ereport(PANIC,
					(errcode(ERRCODE_DATA_CORRUPTED),
					 errmsg("could not read file \"%s\": read %d of %zu",
							fc_path, fc_readBytes, sizeof(fc_magic))));
	}
	COMP_CRC32C(fc_crc, &fc_magic, sizeof(fc_magic));

	if (fc_magic != REPLICATION_STATE_MAGIC)
		ereport(PANIC,
				(errmsg("replication checkpoint has wrong magic %u instead of %u",
						fc_magic, REPLICATION_STATE_MAGIC)));

	/* 我们可以跳过锁定，这里没有其他访问可能 */

	/* 恢复各个状态，直到没有更多可找到的 */
	while (true)
	{
		ReplicationStateOnDisk fc_disk_state;

		fc_readBytes = read(fc_fd, &fc_disk_state, sizeof(fc_disk_state));

		/* 没有更多数据 */
		if (fc_readBytes == sizeof(fc_crc))
		{
			/* 不华丽，但简单... */
			fc_file_crc = *(pg_crc32c *) &fc_disk_state;
			break;
		}

		if (fc_readBytes < 0)
		{
			ereport(PANIC,
					(errcode_for_file_access(),
					 errmsg("could not read file \"%s\": %m",
							fc_path)));
		}

		if (fc_readBytes != sizeof(fc_disk_state))
		{
			ereport(PANIC,
					(errcode_for_file_access(),
					 errmsg("could not read file \"%s\": read %d of %zu",
							fc_path, fc_readBytes, sizeof(fc_disk_state))));
		}

		COMP_CRC32C(fc_crc, &fc_disk_state, sizeof(fc_disk_state));

		if (fc_last_state == max_replication_slots)
			ereport(PANIC,
					(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
					 errmsg("could not find free replication state, increase max_replication_slots")));

		/* 将数据复制到共享内存 */
		replication_states[fc_last_state].roident = fc_disk_state.roident;
		replication_states[fc_last_state].remote_lsn = fc_disk_state.remote_lsn;
		fc_last_state++;

		ereport(LOG,
				(errmsg("recovered replication state of node %d to %X/%X",
						fc_disk_state.roident,
						LSN_FORMAT_ARGS(fc_disk_state.remote_lsn))));
	}

	/* 现在检查校验和 */
	FIN_CRC32C(fc_crc);
	if (fc_file_crc != fc_crc)
		ereport(PANIC,
				(errcode(ERRCODE_DATA_CORRUPTED),
				 errmsg("replication slot checkpoint has wrong checksum %u, expected %u",
						fc_crc, fc_file_crc)));

	if (CloseTransientFile(fc_fd) != 0)
		ereport(PANIC,
				(errcode_for_file_access(),
				 errmsg("could not close file \"%s\": %m",
						fc_path)));
}

void replorigin_redo(XLogReaderState *fc_record)
{
	uint8		fc_info = XLogRecGetInfo(fc_record) & ~XLR_INFO_MASK;

	switch (fc_info)
	{
		case XLOG_REPLORIGIN_SET:
			{
				xl_replorigin_set *fc_xlrec =
				(xl_replorigin_set *) XLogRecGetData(fc_record);

				replorigin_advance(fc_xlrec->node_id,
								   fc_xlrec->remote_lsn, fc_record->EndRecPtr,
								   fc_xlrec->force /* 向后 */ ,
								   false /* WAL 日志 */ );
				break;
			}
		case XLOG_REPLORIGIN_DROP:
			{
				xl_replorigin_drop *fc_xlrec;
				int			fc_i;

				fc_xlrec = (xl_replorigin_drop *) XLogRecGetData(fc_record);

				for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
				{
					ReplicationState *fc_state = &replication_states[fc_i];

					/* 找到我们的插槽 */
					if (fc_state->roident == fc_xlrec->node_id)
					{
						/* 重置条目 */
						fc_state->roident = InvalidRepOriginId;
						fc_state->remote_lsn = InvalidXLogRecPtr;
						fc_state->local_lsn = InvalidXLogRecPtr;
						break;
					}
				}
				break;
			}
		default:
			elog(PANIC, "replorigin_redo: unknown op code %u", fc_info);
	}
}


/*
 * 告诉复制源进度机制，一个来自 'node' 的提交，源自远程节点的 LSN remote_commit，已成功重放，且我们不需要再做一次。结合设置 replorigin_session_origin_lsn 和 replorigin_session_origin，确保我们在崩溃后不会失去关于此的知识，如果事务有持久影响（想想异步提交）。
 *
 * local_commit 必须是提交的本地 LSN，以便我们可以确保在检查点时，足够的 WAL 已经持久化到磁盘。
 *
 * 需要在 pg_replication_origin 上使用 RowExclusiveLock 调用，除非在恢复过程中。
 */
void replorigin_advance(RepOriginId fc_node,
				   XLogRecPtr fc_remote_commit, XLogRecPtr fc_local_commit,
				   bool fc_go_backward, bool fc_wal_log)
{
	int			fc_i;
	ReplicationState *fc_replication_state = NULL;
	ReplicationState *fc_free_state = NULL;

	Assert(fc_node != InvalidRepOriginId);

	/* 我们不跟踪 DoNotReplicateId */
	if (fc_node == DoNotReplicateId)
		return;

	/*
	 * XXX: 对于在 WAL 重放期间调用的情况，将更加高效地恢复到后端本地哈希表中，并在恢复完成后再转储到共享内存中。我们等到证明这是一个可衡量的开销后再实现这一点。
	 */

	/* 独占锁定，因为我们可能需要创建一个新的表条目。 */
	LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);

	/*
	 * 搜索现有的源槽或可以使用的空闲槽。
	 */
	for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
	{
		ReplicationState *fc_curstate = &replication_states[fc_i];

		/* 如果需要，记住在哪里插入 */
		if (fc_curstate->roident == InvalidRepOriginId &&
			fc_free_state == NULL)
		{
			fc_free_state = fc_curstate;
			continue;
		}

		/* 不是我们的槽 */
		if (fc_curstate->roident != fc_node)
		{
			continue;
		}

		/* 好的，找到了槽 */
		fc_replication_state = fc_curstate;

		LWLockAcquire(&fc_replication_state->lock, LW_EXCLUSIVE);

		/* 确保它不被其他人使用 */
		if (fc_replication_state->acquired_by != 0)
		{
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_IN_USE),
					 errmsg("replication origin with ID %d is already active for PID %d",
							fc_replication_state->roident,
							fc_replication_state->acquired_by)));
		}

		break;
	}

	if (fc_replication_state == NULL && fc_free_state == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
				 errmsg("could not find free replication state slot for replication origin with ID %d",
						fc_node),
				 errhint("Increase max_replication_slots and try again.")));

	if (fc_replication_state == NULL)
	{
		/* 初始化新的槽 */
		LWLockAcquire(&fc_free_state->lock, LW_EXCLUSIVE);
		fc_replication_state = fc_free_state;
		Assert(fc_replication_state->remote_lsn == InvalidXLogRecPtr);
		Assert(fc_replication_state->local_lsn == InvalidXLogRecPtr);
		fc_replication_state->roident = fc_node;
	}

	Assert(fc_replication_state->roident != InvalidRepOriginId);

	/*
	 * 如果有人“强制”设置此槽，记入 WAL 日志，以便它是持久的，备用机接收到该消息。主要在 WAL 重放（提交记录）期间调用此项，此时不需要记录 WAL。
	 */
	if (fc_wal_log)
	{
		xl_replorigin_set fc_xlrec;

		fc_xlrec.remote_lsn = fc_remote_commit;
		fc_xlrec.node_id = fc_node;
		fc_xlrec.force = fc_go_backward;

		XLogBeginInsert();
		XLogRegisterData((char *) (&fc_xlrec), sizeof(fc_xlrec));

		XLogInsert(RM_REPLORIGIN_ID, XLOG_REPLORIGIN_SET);
	}

	/*
	 * 由于在检查点期间的 - 无害 - 竞争条件，我们可能会看到这里的值比我们在内存中已有的值更旧。我们也可能会看到准备事务的旧值，当准备在稍后时刻与提交准备一起发送时，并且在准备和提交准备之间有其他事务提交。见 ReorderBufferFinishPrepared。不要覆盖这些。
	 */
	if (fc_go_backward || fc_replication_state->remote_lsn < fc_remote_commit)
		fc_replication_state->remote_lsn = fc_remote_commit;
	if (fc_local_commit != InvalidXLogRecPtr &&
		(fc_go_backward || fc_replication_state->local_lsn < fc_local_commit))
		fc_replication_state->local_lsn = fc_local_commit;
	LWLockRelease(&fc_replication_state->lock);

	/*
	 * 在更改 LSN 后释放 *，槽未被获取，因此可以随时被丢弃。
	 */
	LWLockRelease(ReplicationOriginLock);
}


XLogRecPtr replorigin_get_progress(RepOriginId fc_node, bool fc_flush)
{
	int			fc_i;
	XLogRecPtr	fc_local_lsn = InvalidXLogRecPtr;
	XLogRecPtr	fc_remote_lsn = InvalidXLogRecPtr;

	/* 防止槽被同时丢弃 */
	LWLockAcquire(ReplicationOriginLock, LW_SHARED);

	for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
	{
		ReplicationState *fc_state;

		fc_state = &replication_states[fc_i];

		if (fc_state->roident == fc_node)
		{
			LWLockAcquire(&fc_state->lock, LW_SHARED);

			fc_remote_lsn = fc_state->remote_lsn;
			fc_local_lsn = fc_state->local_lsn;

			LWLockRelease(&fc_state->lock);

			break;
		}
	}

	LWLockRelease(ReplicationOriginLock);

	if (fc_flush && fc_local_lsn != InvalidXLogRecPtr)
		XLogFlush(fc_local_lsn);

	return fc_remote_lsn;
}

/*
 * 在进程退出期间拆除（可能已配置的）会话复制源。
 */
static void fc_ReplicationOriginExitCleanup(int fc_code, Datum fc_arg)
{
	ConditionVariable *fc_cv = NULL;

	LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);

	if (session_replication_state != NULL &&
		session_replication_state->acquired_by == MyProcPid)
	{
		fc_cv = &session_replication_state->origin_cv;

		session_replication_state->acquired_by = 0;
		session_replication_state = NULL;
	}

	LWLockRelease(ReplicationOriginLock);

	if (fc_cv)
		ConditionVariableBroadcast(fc_cv);
}

/*
 * 如果共享内存结构中尚不存在，则在共享内存结构中设置一个复制源，并缓存对特定 ReplicationSlot 的访问，以便在调用 replorigin_session_advance() 时不必搜索数组。
 *
 * 显然，每个进程只能存在一个这样的缓存源，并且当前缓存值只能在用 replorigin_session_reset() 拆除先前值后重新设置。
 */
void replorigin_session_setup(RepOriginId fc_node)
{
	static bool fc_registered_cleanup;
	int			fc_i;
	int			fc_free_slot = -1;

	if (!fc_registered_cleanup)
	{
		on_shmem_exit(fc_ReplicationOriginExitCleanup, 0);
		fc_registered_cleanup = true;
	}

	Assert(max_replication_slots > 0);

	if (session_replication_state != NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("cannot setup replication origin when one is already setup")));

	/* 独占锁定，因为我们可能需要创建一个新的表条目。 */
	LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);

	/*
	 * 搜索现有的源槽或可以使用的空闲槽。
	 */
	for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
	{
		ReplicationState *fc_curstate = &replication_states[fc_i];

		/* 如果需要，记住在哪里插入 */
		if (fc_curstate->roident == InvalidRepOriginId &&
			fc_free_slot == -1)
		{
			fc_free_slot = fc_i;
			continue;
		}

		/* 不是我们的槽 */
		if (fc_curstate->roident != fc_node)
			continue;

		else if (fc_curstate->acquired_by != 0)
		{
			ereport(ERROR,
					(errcode(ERRCODE_OBJECT_IN_USE),
					 errmsg("replication origin with ID %d is already active for PID %d",
							fc_curstate->roident, fc_curstate->acquired_by)));
		}

		/* 好的，找到了槽 */
		session_replication_state = fc_curstate;
	}


	if (session_replication_state == NULL && fc_free_slot == -1)
		ereport(ERROR,
				(errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
				 errmsg("could not find free replication state slot for replication origin with ID %d",
						fc_node),
				 errhint("Increase max_replication_slots and try again.")));
	else if (session_replication_state == NULL)
	{
		/* 初始化新的槽 */
		session_replication_state = &replication_states[fc_free_slot];
		Assert(session_replication_state->remote_lsn == InvalidXLogRecPtr);
		Assert(session_replication_state->local_lsn == InvalidXLogRecPtr);
		session_replication_state->roident = fc_node;
	}


	Assert(session_replication_state->roident != InvalidRepOriginId);

	session_replication_state->acquired_by = MyProcPid;

	LWLockRelease(ReplicationOriginLock);

	/* probably this one is pointless */
	ConditionVariableBroadcast(&session_replication_state->origin_cv);
}

/*
 * 重置在本会话中先前设置的重放状态。
 *
 * 仅当使用 replorigin_session_setup() 设置了源时，可以调用此函数。
 */
void replorigin_session_reset(void)
{
	ConditionVariable *fc_cv;

	Assert(max_replication_slots != 0);

	if (session_replication_state == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("no replication origin is configured")));

	LWLockAcquire(ReplicationOriginLock, LW_EXCLUSIVE);

	session_replication_state->acquired_by = 0;
	fc_cv = &session_replication_state->origin_cv;
	session_replication_state = NULL;

	LWLockRelease(ReplicationOriginLock);

	ConditionVariableBroadcast(fc_cv);
}

/*
 * 执行 replorigin_advance() 所做的相同工作，只是在会话配置的源上。
 *
 * 这比使用 replorigin_advance() 明显便宜。
 */
void replorigin_session_advance(XLogRecPtr fc_remote_commit, XLogRecPtr fc_local_commit)
{
	Assert(session_replication_state != NULL);
	Assert(session_replication_state->roident != InvalidRepOriginId);

	LWLockAcquire(&session_replication_state->lock, LW_EXCLUSIVE);
	if (session_replication_state->local_lsn < fc_local_commit)
		session_replication_state->local_lsn = fc_local_commit;
	if (session_replication_state->remote_lsn < fc_remote_commit)
		session_replication_state->remote_lsn = fc_remote_commit;
	LWLockRelease(&session_replication_state->lock);
}

/*
 * 向机制询问我们成功重放自已设置的复制源的更改的点。
 */
XLogRecPtr replorigin_session_get_progress(bool fc_flush)
{
	XLogRecPtr	fc_remote_lsn;
	XLogRecPtr	fc_local_lsn;

	Assert(session_replication_state != NULL);

	LWLockAcquire(&session_replication_state->lock, LW_SHARED);
	fc_remote_lsn = session_replication_state->remote_lsn;
	fc_local_lsn = session_replication_state->local_lsn;
	LWLockRelease(&session_replication_state->lock);

	if (fc_flush && fc_local_lsn != InvalidXLogRecPtr)
		XLogFlush(fc_local_lsn);

	return fc_remote_lsn;
}



/* ---------------------------------------------------------------------------
 * 用于处理复制源的 SQL 函数。
 *
 * 这些函数大多应该是更通用函数的简短封装。
 * ---------------------------------------------------------------------------
 */

/*
 * 为传入的名称创建复制源，并返回分配的 oid。
 */
Datum pg_replication_origin_create(PG_FUNCTION_ARGS)
{
	char	   *fc_name;
	RepOriginId fc_roident;

	fc_replorigin_check_prerequisites(false, false);

	fc_name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));

	/* 复制源 "pg_xxx" 保留供内部使用 */
	if (IsReservedName(fc_name))
		ereport(ERROR,
				(errcode(ERRCODE_RESERVED_NAME),
				 errmsg("replication origin name \"%s\" is reserved",
						fc_name),
				 errdetail("Origin names starting with \"pg_\" are reserved.")));

	/*
	 * 如果使用了适当的开关，当违反复制源名称的回归测试约定时发出警告。
	 */
#ifdef ENFORCE_REGRESSION_TEST_NAME_RESTRICTIONS
	if (strncmp(fc_name, "regress_", 8) != 0)
		elog(WARNING, "replication origins created by regression test cases should have names starting with \"regress_\"");
#endif

	fc_roident = replorigin_create(fc_name);

	pfree(fc_name);

	PG_RETURN_OID(fc_roident);
}

/*
 * 删除复制源。
 */
Datum pg_replication_origin_drop(PG_FUNCTION_ARGS)
{
	char	   *fc_name;

	fc_replorigin_check_prerequisites(false, false);

	fc_name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));

	replorigin_drop_by_name(fc_name, false, true);

	pfree(fc_name);

	PG_RETURN_VOID();
}

/*
 * 返回复制源的 oid。
 */
Datum pg_replication_origin_oid(PG_FUNCTION_ARGS)
{
	char	   *fc_name;
	RepOriginId fc_roident;

	fc_replorigin_check_prerequisites(false, false);

	fc_name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
	fc_roident = replorigin_by_name(fc_name, true);

	pfree(fc_name);

	if (OidIsValid(fc_roident))
		PG_RETURN_OID(fc_roident);
	PG_RETURN_NULL();
}

/*
 * 为本会话设置一个复制源。
 */
Datum pg_replication_origin_session_setup(PG_FUNCTION_ARGS)
{
	char	   *fc_name;
	RepOriginId fc_origin;

	fc_replorigin_check_prerequisites(true, false);

	fc_name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
	fc_origin = replorigin_by_name(fc_name, false);
	replorigin_session_setup(fc_origin);

	replorigin_session_origin = fc_origin;

	pfree(fc_name);

	PG_RETURN_VOID();
}

/*
 * 重置本会话中先前设置的源。
 */
Datum pg_replication_origin_session_reset(PG_FUNCTION_ARGS)
{
	fc_replorigin_check_prerequisites(true, false);

	replorigin_session_reset();

	replorigin_session_origin = InvalidRepOriginId;
	replorigin_session_origin_lsn = InvalidXLogRecPtr;
	replorigin_session_origin_timestamp = 0;

	PG_RETURN_VOID();
}

/*
 * 本会话是否已设置复制源。
 */
Datum pg_replication_origin_session_is_setup(PG_FUNCTION_ARGS)
{
	fc_replorigin_check_prerequisites(false, false);

	PG_RETURN_BOOL(replorigin_session_origin != InvalidRepOriginId);
}


/*
 * 返回当前会话中设置的源的复制进度。
 *
 * 如果 'flush' 设置为 true，则确保返回的值对应于已刷新（flushed）的本地事务。
 * 这一点在重放复制事务时使用异步提交时非常有用。
 */
Datum pg_replication_origin_session_progress(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_remote_lsn = InvalidXLogRecPtr;
	bool		fc_flush = PG_GETARG_BOOL(0);

	fc_replorigin_check_prerequisites(true, false);

	if (session_replication_state == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("no replication origin is configured")));

	fc_remote_lsn = replorigin_session_get_progress(fc_flush);

	if (fc_remote_lsn == InvalidXLogRecPtr)
		PG_RETURN_NULL();

	PG_RETURN_LSN(fc_remote_lsn);
}

Datum pg_replication_origin_xact_setup(PG_FUNCTION_ARGS)
{
	XLogRecPtr	fc_location = PG_GETARG_LSN(0);

	fc_replorigin_check_prerequisites(true, false);

	if (session_replication_state == NULL)
		ereport(ERROR,
				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
				 errmsg("no replication origin is configured")));

	replorigin_session_origin_lsn = fc_location;
	replorigin_session_origin_timestamp = PG_GETARG_TIMESTAMPTZ(1);

	PG_RETURN_VOID();
}

Datum pg_replication_origin_xact_reset(PG_FUNCTION_ARGS)
{
	fc_replorigin_check_prerequisites(true, false);

	replorigin_session_origin_lsn = InvalidXLogRecPtr;
	replorigin_session_origin_timestamp = 0;

	PG_RETURN_VOID();
}


Datum pg_replication_origin_advance(PG_FUNCTION_ARGS)
{
	text	   *fc_name = PG_GETARG_TEXT_PP(0);
	XLogRecPtr	fc_remote_commit = PG_GETARG_LSN(1);
	RepOriginId fc_node;

	fc_replorigin_check_prerequisites(true, false);

	/* 锁定以防止复制源消失 */
	LockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);

	fc_node = replorigin_by_name(text_to_cstring(fc_name), false);

	/*
	 * 在检查点（checkpoint）时不能合理地传递要刷新的本地提交 - 该事务尚未提交。
	 * 这就是为什么这个函数应该用于设置初始复制状态，而不是用于重放。
	 */
	replorigin_advance(fc_node, fc_remote_commit, InvalidXLogRecPtr,
					   true /* 向后走 */ , true /* WAL 日志 */ );

	UnlockRelationOid(ReplicationOriginRelationId, RowExclusiveLock);

	PG_RETURN_VOID();
}


/*
 * 返回个别复制源的复制进度。
 *
 * 如果 'flush' 设置为 true，则确保返回的值对应于已刷新（flushed）的本地事务。
 * 这一点在重放复制事务时使用异步提交时非常有用。
 */
Datum pg_replication_origin_progress(PG_FUNCTION_ARGS)
{
	char	   *fc_name;
	bool		fc_flush;
	RepOriginId fc_roident;
	XLogRecPtr	fc_remote_lsn = InvalidXLogRecPtr;

	fc_replorigin_check_prerequisites(true, true);

	fc_name = text_to_cstring((text *) DatumGetPointer(PG_GETARG_DATUM(0)));
	fc_flush = PG_GETARG_BOOL(1);

	fc_roident = replorigin_by_name(fc_name, false);
	Assert(OidIsValid(fc_roident));

	fc_remote_lsn = replorigin_get_progress(fc_roident, fc_flush);

	if (fc_remote_lsn == InvalidXLogRecPtr)
		PG_RETURN_NULL();

	PG_RETURN_LSN(fc_remote_lsn);
}


Datum pg_show_replication_origin_status(PG_FUNCTION_ARGS)
{
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	int			fc_i;
#define REPLICATION_ORIGIN_PROGRESS_COLS 4

	/* 如果槽设置为零，我们想返回 0 行 */
	fc_replorigin_check_prerequisites(false, true);

	InitMaterializedSRF(fcinfo, 0);

	/* 防止槽被同时丢弃 */
	LWLockAcquire(ReplicationOriginLock, LW_SHARED);

	/*
	 * 迭代所有可能的 replication_states，显示它们是否被填充。
	 * 注意，我们不进行任何锁定，因此稍微损坏/过期的值是有可能的。
	 */
	for (fc_i = 0; fc_i < max_replication_slots; fc_i++)
	{
		ReplicationState *fc_state;
		Datum		fc_values[REPLICATION_ORIGIN_PROGRESS_COLS];
		bool		fc_nulls[REPLICATION_ORIGIN_PROGRESS_COLS];
		char	   *fc_roname;

		fc_state = &replication_states[fc_i];

		/* 未使用的槽，没有内容可显示 */
		if (fc_state->roident == InvalidRepOriginId)
			continue;

		memset(fc_values, 0, sizeof(fc_values));
		memset(fc_nulls, 1, sizeof(fc_nulls));

		fc_values[0] = ObjectIdGetDatum(fc_state->roident);
		fc_nulls[0] = false;

		/*
		 * 我们没有防止源被并发删除，所以默默接受它可能已经消失。
		 */
		if (replorigin_by_oid(fc_state->roident, true,
							  &fc_roname))
		{
			fc_values[1] = CStringGetTextDatum(fc_roname);
			fc_nulls[1] = false;
		}

		LWLockAcquire(&fc_state->lock, LW_SHARED);

		fc_values[2] = LSNGetDatum(fc_state->remote_lsn);
		fc_nulls[2] = false;

		fc_values[3] = LSNGetDatum(fc_state->local_lsn);
		fc_nulls[3] = false;

		LWLockRelease(&fc_state->lock);

		tuplestore_putvalues(fc_rsinfo->setResult, fc_rsinfo->setDesc,
							 fc_values, fc_nulls);
	}

	LWLockRelease(ReplicationOriginLock);

#undef REPLICATION_ORIGIN_PROGRESS_COLS

	return (Datum) 0;
}
